package org.genitus.druidry

import in.zapr.druid.druidry.{Context, TimeseriesContext}
import in.zapr.druid.druidry.client.{DruidConfiguration, DruidJerseyClient}
import in.zapr.druid.druidry.query.aggregation.{DruidGroupByQuery, DruidTimeSeriesQuery}
import org.slf4j.LoggerFactory

import scala.collection.JavaConversions._

class DruidService(druidConfiguration: DruidConfiguration) {
  private[this] val log = LoggerFactory.getLogger(getClass.getName)

  def groupBySearchFromDruid[T<:DruidGroupBySearch,E<:DruidGroupByEvent](query:T,clazz: Class[E]):List[E] = {
    /** build GroupBy query*/
    val groupByQuery = DruidGroupByQuery.builder()
      .granularity(query.druidGranularity)
      .dataSource(query.druidDataSource)
      .dimensions(query.druidDimensions)
      .intervals(query.druidIntervals)
    query.druidFilter.foreach(x => groupByQuery.filter(x))
    query.druidAggregators.foreach(x => groupByQuery.aggregators(x))
    query.druidPostAggregators.foreach(x => groupByQuery.postAggregators(x))

    /** druid http request */
    val druidJerseyClient = new DruidJerseyClient(druidConfiguration)
    druidJerseyClient.connect()
    val responses = druidJerseyClient.query(groupByQuery.build())
    druidJerseyClient.close()
    log.info(JacksonUtil.toJson(groupByQuery.build()))

    /** json deserialize */
    val javaType = JacksonUtil.javaType(classOf[DruidGroupByResponse[E]],clazz)
    val response = JacksonUtil.parseArray[DruidGroupByResponse[E]](responses,javaType)

    response.map(x => x.getEvent)
  }


  def timeseriesSearchFromDruid[T<:DruidTimeseriesSearch,E<:DruidTimeseriesResult](query:T,clazz: Class[E]):List[E] = {
    /** build GroupBy query*/
    val timeseriesQuery = DruidTimeSeriesQuery.builder()
      .granularity(query.druidGranularity)
      .dataSource(query.druidDataSource)
      .intervals(query.druidIntervals)
      .context(new TimeseriesContext(true))
    query.druidFilter.foreach(x => timeseriesQuery.filter(x))
    query.druidAggregators.foreach(x => timeseriesQuery.aggregators(x))
    query.druidPostAggregators.foreach(x => timeseriesQuery.postAggregators(x))

    /** druid http request */
    val druidJerseyClient = new DruidJerseyClient(druidConfiguration)
    druidJerseyClient.connect()
    val responses = druidJerseyClient.query(timeseriesQuery.build())
    druidJerseyClient.close()
    log.debug(JacksonUtil.toJson(timeseriesQuery.build()))

    /** json deserialize */
    val javaType = JacksonUtil.javaType(classOf[DruidTimeseriesResponse[E]],clazz)
    var response = JacksonUtil.parseArray[DruidTimeseriesResponse[E]](responses,javaType)

    /** empty filling */
    if(query.druidEmptyFilling)
      response = EmptyFilling.timeseriesFill(query.druidIntervals,query.druidGranularity,response)
    response.map(x => x.convertResult(clazz))
  }
}
